package com.bd03.cmccmonitor.homeapp

import com.alibaba.fastjson.JSON
import com.bd03.cmccmonitor.conf.{ConfigHelper, Constant}
import com.bd03.cmccmonitor.dbSql
import com.bd03.cmccmonitor.dbSql.OffsetHandler
import com.bd03.cmccmonitor.util.{CmccKpi, DateUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies}

object CmccMonitorMainApp {
  def main(args: Array[String]): Unit = {
    val ssc = ConfigHelper.getSparkStreamingContext

    val PCODE2PNAME: Broadcast[Map[String, AnyRef]] = ssc.sparkContext.broadcast(ConfigHelper.PCODE2PNAME)



    val cmccMonitorStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](
        ConfigHelper.TOPIC,
        ConfigHelper.KAFKAPARAMS
        //获取初始的偏移量
       // OffsetHandler.obtainOffset4Mysql
      )
    )
    cmccMonitorStream.foreachRDD(rdd=>{
      if(!rdd.isEmpty()) {
        val ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        //业务需求
        //过滤出是充值订单的数据  ,"serviceName"的值是"reChargeNotifyReq"
        val filtered = rdd.map(t =>
          JSON.parseObject(t.value())
        ).filter(_.getString(Constant.SERVICENAME).equals(Constant.SERVICENAME_RCNR))

        val generalData: RDD[(String, String, (Int, Int, Long, Int),String,String)] = filtered.map(t => {
          val bussinessResBoolean = Constant.BUSSINESSRST_SUC.equals(t.getString(Constant.BUSSINESSRST))
          val requestId = t.getString(Constant.REQUESTID) //流水号生成的时间,我们认为是订单的开始时间
          val succCount = if (bussinessResBoolean) 1 else 0 //如果充值成功,订单量+1,不然订单量+0
          val chargefee = if (bussinessResBoolean) t.getString(Constant.CHARGEFEE).toInt else 0 //充值金额,如果是成功的订单,充值金额是获取的值,如果是失败的订单,充值金额给0
          val notifyTime = t.getString(Constant.RECEIVENOTIFYTIME) //20170412030307768   //接收到充值结果的通知时间为订单的结束时间
          val dayHourMinutes: (String, String, String) = DateUtil.getDate(requestId) // 充值成功的时间
          val costTime = DateUtil.consumptionTime(requestId, notifyTime)

          val pCode = t.getString(Constant.PROVINCECODE)

          //当天的日期,当前的小时,(总订单的数量,充值订单的数量,消费时长,消费的金额)
          (dayHourMinutes._1, dayHourMinutes._2, (1, succCount, costTime, chargefee),pCode,dayHourMinutes._3)
          //
        }).cache()

        CmccKpi.generalKpi(generalData)
        CmccKpi.provinceKpi(PCODE2PNAME,generalData)
        CmccKpi.perMinutesKpi(generalData)

        OffsetHandler.savaOffset2Mysql(ranges)
      }
    })

    ssc.start()
    ssc.awaitTermination()
  }

}
